
package com.nvlled.route;

import java.net.*;
import java.util.*;
import java.io.*;
import java.util.concurrent.ConcurrentHashMap;
import javax.swing.JOptionPane;

public class ObjectRouter implements Runnable {
    private static final int BROADCAST_PORT = 7011;
    private static final int BUFFER_SIZE = 1024;

    private AddressTable addressTable;
    private Multimap<String, ObjectReceiver> routes;
    private DatagramSocket socket;
    private IDGenerator idGen;
    private Map<Integer, ORType> receivers;
    private SocketAddress broadcastSocketAddress;
    private InetAddress localAddress;

    public ObjectRouter() throws SocketException, UnknownHostException {
        routes = new Multimap<String, ObjectReceiver>();
        socket = new DatagramSocket();
        addressTable = new AddressTable(new InetSocketAddress(BROADCAST_PORT));
        idGen = new IDGenerator();
        receivers = new ConcurrentHashMap<Integer, ORType>();

        InetAddress address = InetAddress.getByName("255.255.255.255");
        broadcastSocketAddress = new InetSocketAddress(address, BROADCAST_PORT);
        localAddress = findLocalAddress();
    }

    private InetAddress findLocalAddress() throws SocketException, UnknownHostException {
        Enumeration<NetworkInterface> ifaces = NetworkInterface.getNetworkInterfaces();
        if(ifaces.hasMoreElements()) {
            NetworkInterface iface = ifaces.nextElement();
            Enumeration<InetAddress> addresses = iface.getInetAddresses();
            if (addresses.hasMoreElements()) {
                return addresses.nextElement();
            }
        }
        return InetAddress.getByName("localhost");
    }

    public AddressTable getTable() {
        return addressTable;
    }

    public int getLocalPort() {
        return socket.getLocalPort();
    }

    public void purgeRoutes() {
        routes.clear();
        addressTable.clear();
        assert(routes.size() == 0);
        assert(addressTable.size() == 0);
    }

    public InetAddress getLocalInetAddress() {
        return localAddress;
    }

    public InetSocketAddress getLocalAddress() {
        return new InetSocketAddress(localAddress, getLocalPort());
    }


    public void run() {
        try {
            bind();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void bind() throws SocketException {
        DatagramPacket packet = new DatagramPacket(new byte[BUFFER_SIZE], BUFFER_SIZE);

        while (true) {
            try {
                socket.receive(packet);
                RoutedObject obj = fromBytes(packet.getData());
                obj.setSourceAddress(packet.getSocketAddress());

                publish(obj);
            } catch (Exception e) {
                JOptionPane.showMessageDialog(null, "No connection found!");
                System.exit(0);
            }
        }
    }

    public void bindBroadcast() throws SocketException {
        DatagramSocket broadcastSocket = new DatagramSocket(BROADCAST_PORT);
        DatagramPacket packet = new DatagramPacket(new byte[BUFFER_SIZE], BUFFER_SIZE);
        broadcastSocket.setBroadcast(true);
        while (true) {
            try {
                broadcastSocket.receive(packet);
                RoutedObject obj = fromBytes(packet.getData());
                obj.setSourceAddress(packet.getSocketAddress());

                publish(obj);
            } catch (Exception e) {
               JOptionPane.showMessageDialog(null, "No connection found!");
			   System.exit(0);
            }
        }
    }


    public <T extends RoutedObject> T requestOne(
            RoutedObject obj,
            Class<T> expected,
            int duration,
            String name) {


        RequestInfo<T> request 
            = new RequestInfo(expected, 1, duration, false);
        List<T> list = deliverSync(obj, request, name);

        if (list.isEmpty()) {
            return null;
        }
        return list.get(0);
    }


    public <T extends RoutedObject> List<T> deliverSync(
            final RoutedObject obj,
            final RequestInfo<T> req,
            String name) {
        return deliverSync(obj, req, Arrays.asList(name));
    }

    public <T extends RoutedObject> List<T> deliverSync(
            final RoutedObject obj,
            final RequestInfo<T> req,
            Collection<String> names) {

        long id = obj.getId();
        ReplyAccumulator<T> accumulator = new ReplyAccumulator<T>(id, req);
        subscribe(req.getType(), accumulator);

        for (String name: names) {
            deliver(obj, name);
        }
        
        // A subtle bug would be when the thread is untimely interrupted.
        try {
            Thread.sleep(req.getMaxWait());
        } catch (Exception e) { }

        unsubscribe(req.getType(), accumulator);
        return accumulator.getReplies();
    }

    public RoutedObject fromBytes(byte[] data) {
        try {
            ByteArrayInputStream byteStream = new ByteArrayInputStream(data);
            ObjectInputStream objectIn = new ObjectInputStream(byteStream);
            return (RoutedObject) objectIn.readObject();
        } catch (Exception e) {
            return new RoutedObject();
        }
    }

    public byte[] toBytes(Object obj) {
        try {
            ByteArrayOutputStream byteStream = new ByteArrayOutputStream(BUFFER_SIZE);
            ObjectOutputStream objectOut = new ObjectOutputStream(byteStream);
            objectOut.writeObject(obj);
            objectOut.flush();
            return byteStream.toByteArray();
        } catch (Exception e) {
            e.printStackTrace();
            return new byte[] {(byte)0xde, (byte)0xad, (byte)0xbe, (byte)0xef};
        }
    }

    public void deliver(RoutedObject obj, String name) {
        SocketAddress sockAddress = addressTable.getAddress(name);
        if (sockAddress == null) {
            sockAddress = addressTable.getAddress("default");
        }
        deliver(obj, sockAddress);
    }

    
    // TODO: give deliver method a more apt name
    public synchronized void deliver(RoutedObject obj, SocketAddress sockAddress) {
        try {
            byte[] buffer = toBytes(obj);

            DatagramPacket packet = new DatagramPacket(
                    buffer,
                    buffer.length,
                    sockAddress);

            // Note: must be thread-safe
            socket.send(packet);

        } catch (Exception e) {
            //JOptionPane.showMessageDialog(null, "No connection found!");
			//System.exit(0);
        }
    }

    public void deliverBroadcast(RoutedObject obj) {
        deliver(obj, broadcastSocketAddress);
    }

    public void deliver(RoutedObject obj, Collection<String> names) {
        for (String name: names) {
            deliver(obj, name);
        }
    }

    public <T extends RoutedObject> int subscribe(T obj, ObjectReceiver<T> receiver) {
        Class c = obj.getClass();
        return subscribe(c, receiver);
    }

    public <T extends RoutedObject> void unsubscribe(T obj, ObjectReceiver<T> receiver) {
        Class c = obj.getClass();
        unsubscribe(c, receiver);
    }

    public <T extends RoutedObject> int subscribe(Class<T> c, ObjectReceiver<T> receiver) {
        String name = c.getCanonicalName();
        routes.add(name, receiver);
        int id = idGen.nextId();
        receivers.put(id, new ORType(receiver, c));
        return id;
    }

    public void unsubscribe(int id) {
        ORType r = receivers.remove(id);
        if (r != null) {
            unsubscribe(r.type, r.object);
        }
    }

    public <T extends RoutedObject> void unsubscribe(Class<T> c, ObjectReceiver<T> receiver) {
        String name = c.getCanonicalName();
        routes.remove(name, receiver);
    }

    public void publish(RoutedObject obj) {
        Class c = obj.getClass();
        String name = c.getCanonicalName();

        Collection<ObjectReceiver> ors = routes.getValues(name);
        if (ors == null) {
            return;
        }

        // TODO: Use a thread pool
        for(ObjectReceiver or: ors) { 
            new Thread(new Dispatcher(or, obj)).start();
        }
    }

    class Dispatcher implements Runnable {
        ObjectReceiver receiver;
        RoutedObject object;
        public Dispatcher(ObjectReceiver or, RoutedObject obj) {
            receiver = or;
            object = obj;
        }

        public void run() {
            receiver.receive(object);
        }
    }


    private class ORType {

        ObjectReceiver object;
        Class type;

        public ORType(ObjectReceiver obj, Class t) {
            object = obj;
            type = t;
        }
    }

    private class ReplyAccumulator<T extends RoutedObject> implements ObjectReceiver<T> {

        private RequestInfo request;
        private List<T> replies;
        private Thread waitThread;
        private long requestId;

        public ReplyAccumulator(long id, RequestInfo req) {
            requestId = id;
            request = req;
            replies = new ArrayList<T>();
            waitThread = Thread.currentThread();
        }

        public List<T> getReplies() {
            return replies;
        }

        public void receive(T receivedObj) {
            if (!request.isById() || receivedObj.getId() == requestId) {
                // Note: possible race condition
                replies.add(receivedObj);
            }
            if (replies.size() >= request.getNumReplies()) {
                waitThread.interrupt();
            }
        }
    }
}

